package com.hxq.demo6;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.SaslConfigs;

import java.time.Duration;
import java.util.*;

/**自动提交：消费者可能会丢数据，如：消费到100条数据，自动提交时间是1秒，offset被提交了，但是还没有消费完数据，此时消费者挂掉了。当消费者再次启动，会从101继续消费
 * 手动提交：
 * 虽然自动提交 offset 十分简介便利，但由于其是基于时间提交的，开发人员难以把握
 * offset 提交的时机。因此 Kafka 还提供了手动提交 offset 的 API。
 * 手动提交 offset 的方法有两种：分别是 commitSync（同步提交）和 commitAsync（异步
 * 提交）。两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交；不同点是，
 * commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，
 * 也会出现提交失败）；而 commitAsync 则没有失败重试机制，故有可能提交失败。
 * @author hxq
 * @date 2022/4/14 18:08
 */
public class SASLConsumer {

    public static void main(String[] args) {
        autoCommitOffset();
    }

    private static void autoCommitOffset(){

        Properties properties = setConfigInfo(true);
//        Properties properties = setConfigInfo(false);//如果不开启自动提交，也不设置手动提交，则不会提交offset,所以每次启动消费者都会消费到上一个offset的数据
        //创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        consumer.subscribe(Arrays.asList("test"));//消费者订阅主题

        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records){
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }

        }

    }





    private static Properties setConfigInfo(Boolean isAuto){
        Properties props = new Properties();
        props.put("bootstrap.servers", "VM-12-12-ubuntu:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", isAuto);//是否开启自动提交 offset 功能
        props.put("auto.commit.interval.ms", "1000");//自动提交 offset 的时间间隔
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        //设置SASL连接
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
        //方式一
        props.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule required username=\"reader\" password=\"123456\";");
        return props;
    }
}
